package com.duowan.realtime.scheduled.batch.reader.hyperLog;


import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.item.*;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;

import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * 从hyerlog中持续读取数据，sleep_second指定了读取两批数据之间的间隔，默认10秒
 *
 * @author tangsicheng
 * @version 1.0
 * @since 1.0
 */
public class HyerLogContinueReader extends HyperLogReader implements ItemStreamReader<List<Map>>, InitializingBean {

    private static final Logger logger = LoggerFactory.getLogger(HyerLogContinueReader.class);

    private String kpiCode;
    private String hyperLogGroup;
    private int sleep_second = DEFAULT_SLEEP_SECOND;
    private boolean firstRun = true;

    private static int DEFAULT_SLEEP_SECOND = 10;

    @Override
    public void afterPropertiesSet() throws Exception {
        Assert.hasText(kpiCode, "kpiCode is empty");
        Assert.hasText(hyperLogGroup, "hyperLogGroup is empty");
    }

    @Override
    public List<Map> read() throws UnexpectedInputException, ParseException, NonTransientResourceException, InterruptedException {
        if (!firstRun) {
            if (sleep_second > 0) {
                logger.info(" sleep " + sleep_second + "second");
                TimeUnit.SECONDS.sleep(sleep_second);
            }
        }
        firstRun = false;
        return readFromHyperLog(hyperLogGroup, kpiCode);
    }

    @Override
    public void open(ExecutionContext executionContext) throws ItemStreamException {

    }

    @Override
    public void update(ExecutionContext executionContext) throws ItemStreamException {

    }

    @Override
    public void close() throws ItemStreamException {

    }

    public void setKpiCode(String kpiCode) {
        this.kpiCode = kpiCode;
    }

    public void setHyperLogGroup(String hyperLogGroup) {
        this.hyperLogGroup = hyperLogGroup;
    }

    public void setSleep_second(int sleep_second) {
        this.sleep_second = sleep_second;
    }
}
